package main.scala.com.hu.api

import com.hu.entity.SensorReading
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.scala._

/**
 * @Author: 22834
 * @Date: 2020/11/19 22:21
 * @Describe:
 */
object StateDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val fsBackend = new FsStateBackend("file:///D:/project/java8/data/fsdb",false)
    env.setStateBackend(fsBackend)

    //使用rocksdb保存
    val rocksBackend = new RocksDBStateBackend("file:///D:/project/java8/data/rocksdb", true).getCheckpointBackend
    env.setStateBackend(rocksBackend)

    val socketStream = env.socketTextStream("localhost", 7777)

    val sensorStream = socketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    sensorStream.keyBy(_.id).flatMapWithState[(String, Double, Double), Double]({
      case (data: SensorReading, None) => (List.empty, Some(data.temperature))
      case (data: SensorReading, lastTemp: Some[Double]) => {
        val diff = (data.temperature - lastTemp.get).abs
        if (diff > 10.0) (List((data.id, lastTemp.get, data.temperature)), Some(data.temperature))
        else (List.empty, Some(data.temperature))
      }
    }).print()

    env.execute("测试状态编程")
  }
}
